<!DOCTYPE html>
<html lang="en">
<head>
  <title>Vert.x RxJava - Vert.x</title>
  <meta charset="utf-8">
  <meta http-equiv="X-UA-Compatible" content="IE=edge">
  <meta content="width=device-width, initial-scale=1.0" name="viewport">
  <meta content="Vert.x is a tool-kit for building reactive applications on the JVM." name="description">
  <link href="http://vertx.io/stylesheets/docs.css" media="screen" rel="stylesheet">
  <link href="http://vertx.io/stylesheets/font-awesome.min.css" media="screen" rel="stylesheet">
  <link href="http://vertx.io/javascripts/styles/rainbow.min.css" media="screen" rel="stylesheet">
  <!-- IE 6-8 support of HTML 5 elements -->
  <!--[if lt IE 9]>
  <script src="http://static.jboss.org/theme/js/libs/html5/pre3.6/html5.min.js"></script>
  <![endif]-->

  <link rel="apple-touch-icon" sizes="57x57" href="http://vertx.io/assets/favicons/vertx-favicon-7/apple-touch-icon-57x57.png">
  <link rel="apple-touch-icon" sizes="60x60" href="http://vertx.io/assets/favicons/vertx-favicon-7/apple-touch-icon-60x60.png">
  <link rel="apple-touch-icon" sizes="72x72" href="http://vertx.io/assets/favicons/vertx-favicon-7/apple-touch-icon-72x72.png">
  <link rel="apple-touch-icon" sizes="76x76" href="http://vertx.io/assets/favicons/vertx-favicon-7/apple-touch-icon-76x76.png">
  <link rel="apple-touch-icon" sizes="114x114" href="http://vertx.io/assets/favicons/vertx-favicon-7/apple-touch-icon-114x114.png">
  <link rel="apple-touch-icon" sizes="120x120" href="http://vertx.io/assets/favicons/vertx-favicon-7/apple-touch-icon-120x120.png">
  <link rel="apple-touch-icon" sizes="144x144" href="http://vertx.io/assets/favicons/vertx-favicon-7/apple-touch-icon-144x144.png">
  <link rel="apple-touch-icon" sizes="152x152" href="http://vertx.io/assets/favicons/vertx-favicon-7/apple-touch-icon-152x152.png">
  <link rel="apple-touch-icon" sizes="180x180" href="http://vertx.io/assets/favicons/vertx-favicon-7/apple-touch-icon-180x180.png">
  <link rel="icon" type="image/png" href="http://vertx.io/assets/favicons/vertx-favicon-7/favicon-32x32.png" sizes="32x32">
  <link rel="icon" type="image/png" href="http://vertx.io/assets/favicons/vertx-favicon-7/android-chrome-192x192.png" sizes="192x192">
  <link rel="icon" type="image/png" href="http://vertx.io/assets/favicons/vertx-favicon-7/favicon-96x96.png" sizes="96x96">
  <link rel="icon" type="image/png" href="http://vertx.io/assets/favicons/vertx-favicon-7/favicon-16x16.png" sizes="16x16">
  <link rel="manifest" href="http://vertx.io/assets/favicons/vertx-favicon-7/manifest.json">
  <link rel="mask-icon" href="http://vertx.io/assets/favicons/vertx-favicon-7/safari-pinned-tab.svg" color="#5bbad5">
  <meta name="msapplication-TileColor" content="#7d3194">
  <meta name="msapplication-TileImage" content="http://vertx.io/assets/favicons/vertx-favicon-7/mstile-144x144.png">
  <meta name="theme-color" content="#ffffff">

  <link href="http://fonts.googleapis.com/css?family=Ubuntu:400,500,700,400italic" rel="stylesheet" type="text/css">
  <link rel="alternate" type="application/rss+xml" title="RSS"
     href="http://vertx.io/feed.xml">
  <script>
    (function(i,s,o,g,r,a,m){i['GoogleAnalyticsObject']=r;i[r]=i[r]||function(){
      (i[r].q=i[r].q||[]).push(arguments)},i[r].l=1*new Date();a=s.createElement(o),
      m=s.getElementsByTagName(o)[0];a.async=1;a.src=g;m.parentNode.insertBefore(a,m)
    })(window,document,'script','//www.google-analytics.com/analytics.js','ga');
    ga('create', 'UA-30144458-1', 'auto');
    ga('create', 'UA-71153120-1', 'auto', 'tracker');
    ga('send', 'pageview');
    ga('tracker.send', 'pageview');
  </script>
</head>
<body>

<a href="http://www.reactivemanifesto.org/" id="reactive-manifesto-banner">
  <img style="border: 0; position: fixed; right: 0; top:0; z-index: 9000"
    src="http://d379ifj7s9wntv.cloudfront.net/reactivemanifesto/images/ribbons/we-are-reactive-black-right.png">
</a>

<a id="skippy" class="sr-only sr-only-focusable" href="#content"><div class="container"><span class="skiplink-text">Skip to main content</span></div></a>

<header class="navbar navbar-default navbar-static-top" id="top" role="banner">
  <div class="container">
    <div class="navbar-header">
      <button class="navbar-toggle collapsed" type="button" data-toggle="collapse" data-target="#vertx-navbar-collapse">
        <span class="sr-only">Toggle navigation</span>
        <span class="icon-bar"></span>
        <span class="icon-bar"></span>
        <span class="icon-bar"></span>
      </button>
      <a href="http://vertx.io/" class="navbar-brand"><img alt="Brand" src="http://vertx.io/assets/logo-sm.png"></a>
    </div>
    <nav class="collapse navbar-collapse" id="vertx-navbar-collapse">
      <ul class="nav navbar-nav navbar-right">
        <li><a href="http://vertx.io/download/">Download</a></li>
        <li><a href="http://vertx.io/docs/">Documentation</a></li>
        <li><a href="https://github.com/vert-x3/wiki/wiki">Wiki</a></li>
        <li><a href="http://vertx.io/community/">Community</a></li>
        <li><a href="http://vertx.io/materials/">Materials</a></li>
        <li><a href="http://vertx.io/blog/">Blog</a></li>        
      </ul>
    </nav>
  </div>
</header>



  <div class="page-header" id="content">
    <div class="container">
      <div class="row">
        <div class="col-sm-12">
          <h1>Vert.x RxJava</h1>
          
        </div>
      </div>
    </div>
  </div>




<div id="content">
  <div class="container docs-content">
    <div class="row">
      <div class="col-sm-12 col-md-push-9 col-md-3 hidden-xs hidden-sm">
        <div id="sidebar" data-spy="affix">
          <ul class="sectlevel1">
<li><a href="#_vert_x_api_for_rxjava">Vert.x API for RxJava</a>
<ul class="sectlevel2">
<li><a href="#_read_stream_support">Read stream support</a></li>
<li><a href="#_handler_support">Handler support</a></li>
<li><a href="#_async_result_support">Async result support</a></li>
<li><a href="#_scheduler_support">Scheduler support</a></li>
<li><a href="#_json_unmarshalling">Json unmarshalling</a></li>
<li><a href="#_deploying_a_verticle">Deploying a Verticle</a></li>
<li><a href="#_httpclient_get_on_subscription">HttpClient GET on subscription</a></li>
</ul>
</li>
<li><a href="#_rxified_api">Rxified API</a>
<ul class="sectlevel2">
<li><a href="#_embedding_rxfified_vert_x">Embedding Rxfified Vert.x</a></li>
<li><a href="#_as_a_verticle">As a Verticle</a></li>
<li><a href="#_api_examples">Api examples</a>
<ul class="sectlevel2">
<li><a href="#_eventbus_message_stream">EventBus message stream</a></li>
<li><a href="#_timers">Timers</a></li>
<li><a href="#_http_client_requests">Http client requests</a></li>
<li><a href="#_http_server_requests">Http server requests</a></li>
<li><a href="#_websocket_client">Websocket client</a></li>
<li><a href="#_websocket_server">Websocket server</a></li>
</ul>
</li>
</ul>
</li>
</ul>
        </div>
      </div>
      <div class="col-sm-12 col-md-pull-3 col-md-9">
        <div class="toc hidden-md hidden-lg">
          <h2>Table of Contents</h2>
          <ul class="sectlevel1">
<li><a href="#_vert_x_api_for_rxjava">Vert.x API for RxJava</a>
<ul class="sectlevel2">
<li><a href="#_read_stream_support">Read stream support</a></li>
<li><a href="#_handler_support">Handler support</a></li>
<li><a href="#_async_result_support">Async result support</a></li>
<li><a href="#_scheduler_support">Scheduler support</a></li>
<li><a href="#_json_unmarshalling">Json unmarshalling</a></li>
<li><a href="#_deploying_a_verticle">Deploying a Verticle</a></li>
<li><a href="#_httpclient_get_on_subscription">HttpClient GET on subscription</a></li>
</ul>
</li>
<li><a href="#_rxified_api">Rxified API</a>
<ul class="sectlevel2">
<li><a href="#_embedding_rxfified_vert_x">Embedding Rxfified Vert.x</a></li>
<li><a href="#_as_a_verticle">As a Verticle</a></li>
<li><a href="#_api_examples">Api examples</a>
<ul class="sectlevel2">
<li><a href="#_eventbus_message_stream">EventBus message stream</a></li>
<li><a href="#_timers">Timers</a></li>
<li><a href="#_http_client_requests">Http client requests</a></li>
<li><a href="#_http_server_requests">Http server requests</a></li>
<li><a href="#_websocket_client">Websocket client</a></li>
<li><a href="#_websocket_server">Websocket server</a></li>
</ul>
</li>
</ul>
</li>
</ul>
        </div>
        <div class="sect1">
<h2 id="_vert_x_api_for_rxjava">Vert.x API for RxJava</h2>
<div class="sectionbody">
<div class="paragraph">
<p><a href="https://github.com/ReactiveX/RxJava">RxJava</a> is a popular library for composing asynchronous and event based programs using
observable sequences for the Java VM.</p>
</div>
<div class="paragraph">
<p>Vert.x integrates naturally with RxJava, allowing to use
observable wherever you can use streams or asynchronous results.</p>
</div>
<div class="paragraph">
<p>There are two ways for using the RxJava API with Vert.x:</p>
</div>
<div class="ulist">
<ul>
<li>
<p>via the original Vert.x API with the <code><a href="../../apidocs/io/vertx/rx/java/RxHelper.html">RxHelper</a></code> helper class that provides static
methods for converting objects between Vert.x core API and RxJava API.</p>
</li>
<li>
<p>via the <em>Rxified</em> Vert.x API enhancing the core Vert.x API.</p>
</li>
</ul>
</div>
<div class="sect2">
<h3 id="_read_stream_support">Read stream support</h3>
<div class="paragraph">
<p>RxJava observable is a perfect match for Vert.x <code>ReadStream</code> class : both provide a flow of items.</p>
</div>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rx/java/RxHelper.html#toObservable-io.vertx.core.streams.ReadStream-">RxHelper.toObservable</a></code> static methods convert
a Vert.x read stream to an <code>rx.Observable</code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">FileSystem fileSystem = vertx.fileSystem();
fileSystem.open("/data.txt", new OpenOptions(), result -&gt; {
  AsyncFile file = result.result();
  Observable&lt;Buffer&gt; observable = RxHelper.toObservable(file);
  observable.forEach(data -&gt; System.out.println("Read data: " + data.toString("UTF-8")));
});</code></pre>
</div>
</div>
<div class="paragraph">
<p>The <em>Rxified</em> Vert.x API provides a <code><a href="../../apidocs/io/vertx/rxjava/core/streams/ReadStream.html#toObservable--">toObservable</a></code>  method on
<code><a href="../../apidocs/io/vertx/rxjava/core/streams/ReadStream.html">ReadStream</a></code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">FileSystem fs = vertx.fileSystem();
fs.open("/data.txt", new OpenOptions(), result -&gt; {
  AsyncFile file = result.result();
  Observable&lt;Buffer&gt; observable = file.toObservable();
  observable.forEach(data -&gt; System.out.println("Read data: " + data.toString("UTF-8")));
});</code></pre>
</div>
</div>
<div class="paragraph">
<p>Such observables are <strong>hot</strong> observables, i.e. they will produce notifications regardless of subscriptions.</p>
</div>
</div>
<div class="sect2">
<h3 id="_handler_support">Handler support</h3>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rx/java/RxHelper.html">RxHelper</a></code> can create an <code><a href="../../apidocs/io/vertx/rx/java/ObservableHandler.html">ObservableHandler</a></code>: an <code>Observable</code> with a
<code><a href="../../apidocs/io/vertx/rx/java/ObservableHandler.html#toHandler--">toHandler</a></code> method returning an <code>Handler&lt;T&gt;</code> implementation:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">ObservableHandler&lt;Long&gt; observable = RxHelper.observableHandler();
observable.subscribe(id -&gt; {
  // Fired
});
vertx.setTimer(1000, observable.toHandler());</code></pre>
</div>
</div>
<div class="paragraph">
<p>The <em>Rxified</em> Vert.x API does not provide a specific API for handler.</p>
</div>
</div>
<div class="sect2">
<h3 id="_async_result_support">Async result support</h3>
<div class="paragraph">
<p>The Vert.x <code>Handler&lt;AsyncResult&lt;T&gt;&gt;</code> construct occuring as last parameter of an asynchronous method can
be mapped to an observable of a single element:</p>
</div>
<div class="ulist">
<ul>
<li>
<p>when the callback is a success, the observer <code>onNext</code> method is called with the item
and the <code>onComplete</code> method is immediately invoked after</p>
</li>
<li>
<p>when the callback is a failure, the observer <code>onError</code> method is called</p>
</li>
</ul>
</div>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rx/java/RxHelper.html#observableFuture--">RxHelper.observableFuture</a></code> method creates an <code><a href="../../apidocs/io/vertx/rx/java/ObservableFuture.html">ObservableFuture</a></code>:
an <code>Observable</code> with a <code><a href="../../apidocs/io/vertx/rx/java/ObservableFuture.html#toHandler--">toHandler</a></code> method returning a <code>Handler&lt;AsyncResult&lt;T&gt;&gt;</code>
implementation:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">ObservableFuture&lt;HttpServer&gt; observable = RxHelper.observableFuture();
observable.subscribe(
    server -&gt; {
      // Server is listening
    },
    failure -&gt; {
      // Server could not start
    }
);
vertx.createHttpServer(new HttpServerOptions().
    setPort(1234).
    setHost("localhost")
).listen(observable.toHandler());</code></pre>
</div>
</div>
<div class="paragraph">
<p>The <code>ObservableFuture&lt;Server&gt;</code> will get a single <code>HttpServer</code> object, if the <code>listen</code> operation fails,
the subscriber will be notified with the failure.</p>
</div>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rx/java/RxHelper.html#toHandler-rx.Observer-">RxHelper.toHandler</a></code> method adapts an existing <code>Observer</code> into an handler:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">Observer&lt;HttpServer&gt; observer = new Observer&lt;HttpServer&gt;() {
  @Override
  public void onNext(HttpServer o) {
  }
  @Override
  public void onError(Throwable e) {
  }
  @Override
  public void onCompleted() {
  }
};
Handler&lt;AsyncResult&lt;HttpServer&gt;&gt; handler = RxHelper.toFuture(observer);</code></pre>
</div>
</div>
<div class="paragraph">
<p>It also works with just actions:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">Action1&lt;HttpServer&gt; onNext = httpServer -&gt; {};
Action1&lt;Throwable&gt; onError = httpServer -&gt; {};
Action0 onComplete = () -&gt; {};

Handler&lt;AsyncResult&lt;HttpServer&gt;&gt; handler1 = RxHelper.toFuture(onNext);
Handler&lt;AsyncResult&lt;HttpServer&gt;&gt; handler2 = RxHelper.toFuture(onNext, onError);
Handler&lt;AsyncResult&lt;HttpServer&gt;&gt; handler3 = RxHelper.toFuture(onNext, onError, onComplete);</code></pre>
</div>
</div>
<div class="paragraph">
<p>The <em>Rxified</em> Vert.x API duplicates each such method with the <code>Observable</code> suffix that returns an observable:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">vertx.createHttpServer(
    new HttpServerOptions().setPort(1234).setHost("localhost")
).listenObservable().
    subscribe(
        server -&gt; {
          // Server is listening
        },
        failure -&gt; {
          // Server could not start
        }
    );</code></pre>
</div>
</div>
<div class="paragraph">
<p>Such observables are <strong>cold</strong> observables, i.e. they will produce notifications on request.</p>
</div>
</div>
<div class="sect2">
<h3 id="_scheduler_support">Scheduler support</h3>
<div class="paragraph">
<p>The reactive extension sometimes needs to schedule actions, for instance <code>Observable#timer</code> creates and returns
a timer that emit periodic events. By default, scheduled actions are managed by RxJava, it means that the
timer threads are not Vert.x threads and therefore not executing in a Vert.x event loop.</p>
</div>
<div class="paragraph">
<p>When an RxJava method deals with a scheduler, it accepts an overloaded method accepting an extra <code>rx.Scheduler</code>,
the <code><a href="../../apidocs/io/vertx/rx/java/RxHelper.html#scheduler-io.vertx.core.Vertx-">RxHelper.scheduler</a></code>  method will return a scheduler that can be used
in such places.</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">Scheduler scheduler = RxHelper.scheduler(vertx);
Observable&lt;Long&gt; timer = Observable.timer(100, 100, TimeUnit.MILLISECONDS, scheduler);</code></pre>
</div>
</div>
<div class="paragraph">
<p>For blocking scheduled actions, a scheduler can be created with the <code><a href="../../apidocs/io/vertx/rx/java/RxHelper.html#blockingScheduler-io.vertx.core.Vertx-">RxHelper.blockingScheduler</a></code>
method:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">Scheduler scheduler = RxHelper.blockingScheduler(vertx);
Observable&lt;Integer&gt; obs = blockingObservable.observeOn(scheduler);</code></pre>
</div>
</div>
<div class="paragraph">
<p>RxJava can also be reconfigured to use the Vert.x scheduler, thanks to the scheduler hook created with
<code><a href="../../apidocs/io/vertx/rx/java/RxHelper.html#schedulerHook-io.vertx.core.Vertx-">RxHelper.schedulerHook</a></code>, the returned scheduler hook
uses a blocking scheduler for IO actions:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">RxJavaSchedulersHook hook = RxHelper.schedulerHook(vertx);
rx.plugins.RxJavaPlugins.getInstance().registerSchedulersHook(hook);</code></pre>
</div>
</div>
<div class="paragraph">
<p>The <em>Rxified</em> Vert.x API provides also similar method on the <code><a href="../../apidocs/io/vertx/rxjava/core/RxHelper.html">RxHelper</a></code> class:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">Scheduler scheduler = io.vertx.rxjava.core.RxHelper.scheduler(vertx);
Observable&lt;Long&gt; timer = Observable.timer(100, 100, TimeUnit.MILLISECONDS, scheduler);</code></pre>
</div>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">RxJavaSchedulersHook hook = io.vertx.rxjava.core.RxHelper.schedulerHook(vertx);
rx.plugins.RxJavaPlugins.getInstance().registerSchedulersHook(hook);</code></pre>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_json_unmarshalling">Json unmarshalling</h3>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rxjava/core/RxHelper.html#unmarshaller-java.lang.Class-">RxHelper.unmarshaller</a></code> creates an <code>rx.Observable.Operator</code> that
transforms an <code>Observable&lt;Buffer&gt;</code> in json format into an object observable:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">fileSystem.open("/data.txt", new OpenOptions(), result -&gt; {
  AsyncFile file = result.result();
  Observable&lt;Buffer&gt; observable = RxHelper.toObservable(file);
  observable.lift(RxHelper.unmarshaller(MyPojo.class)).subscribe(
      mypojo -&gt; {
        // Process the object
      }
  );
});</code></pre>
</div>
</div>
<div class="paragraph">
<p>The same can be done with the <em>Rxified</em> helper:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">fileSystem.open("/data.txt", new OpenOptions(), result -&gt; {
  AsyncFile file = result.result();
  Observable&lt;Buffer&gt; observable = file.toObservable();
  observable.lift(io.vertx.rxjava.core.RxHelper.unmarshaller(MyPojo.class)).subscribe(
      mypojo -&gt; {
        // Process the object
      }
  );
});</code></pre>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_deploying_a_verticle">Deploying a Verticle</h3>
<div class="paragraph">
<p>The Rxified API cannot deploy an existing Verticle instance, the helper <code><a href="../../apidocs/io/vertx/rx/java/RxHelper.html#observableFuture--">RxHelper.observableFuture</a></code> method
provides a solution to that.</p>
</div>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rxjava/core/RxHelper.html#deployVerticle-io.vertx.rxjava.core.Vertx-io.vertx.core.Verticle-">RxHelper.deployVerticle</a></code> does it automatically
for you, it deploys a <code>Verticle</code> and returns an <code>Observable&lt;String&gt;</code> of the deployment ID.</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">Observable&lt;String&gt; deployment = RxHelper.deployVerticle(vertx, verticle);

deployment.subscribe(id -&gt; {
  // Deployed
}, err -&gt; {
  // Could not deploy
});</code></pre>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_httpclient_get_on_subscription">HttpClient GET on subscription</h3>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rxjava/core/RxHelper.html#get-io.vertx.rxjava.core.http.HttpClient-int-java.lang.String-java.lang.String-">RxHelper.get</a></code>
is a convenient helper method that performs an HTTP GET upon subscription:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">Observable&lt;HttpClientResponse&gt; get = RxHelper.get(client, "http://the-server");

// Do the request
get.subscribe(resp -&gt; {
  // Got response
}, err -&gt; {
  // Something went wrong
});</code></pre>
</div>
</div>
<div class="admonitionblock warning">
<table>
<tr>
<td class="icon">
<div class="title">Warning</div>
</td>
<td class="content">
this API is different from the HttpClient that performs the GET request when the method is called and returns
a one shot <code>Observable</code>.
</td>
</tr>
</table>
</div>
</div>
</div>
</div>
<h1 id="_rxified_api" class="sect0">Rxified API</h1>
<div class="openblock partintro">
<div class="content">
The <em>Rxified</em> API is a code generated version of the Vert.x API, just like the <em>JavaScript</em> or <em>Groovy</em>
language. The API uses the <code>io.vertx.rxjava</code> prefix, for instance the <code>io.vertx.core.Vertx</code> class is
translated to the <code><a href="../../apidocs/io/vertx/rxjava/core/Vertx.html">Vertx</a></code> class.
</div>
</div>
<div class="sect2">
<h3 id="_embedding_rxfified_vert_x">Embedding Rxfified Vert.x</h3>
<div class="paragraph">
<p>Just use the <code><a href="../../apidocs/io/vertx/rxjava/core/Vertx.html#vertx--">Vertx.vertx</a></code> methods:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">Vertx vertx = io.vertx.rxjava.core.Vertx.vertx();</code></pre>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_as_a_verticle">As a Verticle</h3>
<div class="paragraph">
<p>Extend the <code><a href="../../apidocs/io/vertx/rxjava/core/AbstractVerticle.html">AbstractVerticle</a></code> class, it will wrap it for you:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">class MyVerticle extends io.vertx.rxjava.core.AbstractVerticle {
  public void start() {
    // Use Rxified Vertx here
  }
}</code></pre>
</div>
</div>
<div class="paragraph">
<p>Deploying an RxJava verticle is still performed by the Java deployer and does not need a specified
deployer.</p>
</div>
</div>
<div class="sect1">
<h2 id="_api_examples">Api examples</h2>
<div class="sectionbody">
<div class="paragraph">
<p>Let&#8217;s study now a few examples of using Vert.x with RxJava.</p>
</div>
<div class="sect2">
<h3 id="_eventbus_message_stream">EventBus message stream</h3>
<div class="paragraph">
<p>The event bus <code><a href="../../apidocs/io/vertx/rxjava/core/eventbus/MessageConsumer.html">MessageConsumer</a></code> provides naturally an <code>Observable&lt;Message&lt;T&gt;&gt;</code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">EventBus eb = vertx.eventBus();
MessageConsumer&lt;String&gt; consumer = eb.&lt;String&gt;consumer("the-address");
Observable&lt;Message&lt;String&gt;&gt; observable = consumer.toObservable();
Subscription sub = observable.subscribe(msg -&gt; {
  // Got message
});

// Unregisters the stream after 10 seconds
vertx.setTimer(10000, id -&gt; {
  sub.unsubscribe();
});</code></pre>
</div>
</div>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rxjava/core/eventbus/MessageConsumer.html">MessageConsumer</a></code> provides a stream of <code><a href="../../apidocs/io/vertx/rxjava/core/eventbus/Message.html">Message</a></code>.
The <code><a href="../../apidocs/io/vertx/rxjava/core/eventbus/Message.html#body--">body</a></code> gives access to a new stream of message bodies if needed:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">EventBus eb = vertx.eventBus();
MessageConsumer&lt;String&gt; consumer = eb.&lt;String&gt;consumer("the-address");
Observable&lt;String&gt; observable = consumer.bodyStream().toObservable();</code></pre>
</div>
</div>
<div class="paragraph">
<p>RxJava map/reduce composition style can then be used:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">Observable&lt;Double&gt; observable = vertx.eventBus().
    &lt;Double&gt;consumer("heat-sensor").
    bodyStream().
    toObservable();

observable.
    buffer(1, TimeUnit.SECONDS).
    map(samples -&gt; samples.
        stream().
        collect(Collectors.averagingDouble(d -&gt; d))).
    subscribe(heat -&gt; {
      vertx.eventBus().send("news-feed", "Current heat is " + heat);
    });</code></pre>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_timers">Timers</h3>
<div class="paragraph">
<p>Timer task can be created with <code><a href="../../apidocs/io/vertx/rxjava/core/Vertx.html#timerStream-long-">timerStream</a></code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">vertx.timerStream(1000).
    toObservable().
    subscribe(
        id -&gt; {
          System.out.println("Callback after 1 second");
        }
    );</code></pre>
</div>
</div>
<div class="paragraph">
<p>Periodic task can be created with <code><a href="../../apidocs/io/vertx/rxjava/core/Vertx.html#periodicStream-long-">periodicStream</a></code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">vertx.periodicStream(1000).
    toObservable().
    subscribe(
        id -&gt; {
          System.out.println("Callback every second");
        }
    );</code></pre>
</div>
</div>
<div class="paragraph">
<p>The observable can be cancelled with an unsubscription:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">vertx.periodicStream(1000).
    toObservable().
    subscribe(new Subscriber&lt;Long&gt;() {
      public void onNext(Long aLong) {
        // Callback
        unsubscribe();
      }
      public void onError(Throwable e) {}
      public void onCompleted() {}
    });</code></pre>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_http_client_requests">Http client requests</h3>
<div class="paragraph">
<p><code><a href="../../apidocs/io/vertx/rxjava/core/http/HttpClientRequest.html#toObservable--">toObservable</a></code> provides a one shot callback with the
<code><a href="../../apidocs/io/vertx/core/http/HttpClientResponse.html">HttpClientResponse</a></code> object. The observable reports a request failure.</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">HttpClient client = vertx.createHttpClient(new HttpClientOptions());
HttpClientRequest request = client.request(HttpMethod.GET, 8080, "localhost", "/the_uri");
request.toObservable().subscribe(
    response -&gt; {
      // Process the response
    },
    error -&gt; {
      // Could not connect
    }
);
request.end();</code></pre>
</div>
</div>
<div class="paragraph">
<p>The response can be processed as an <code>Observable&lt;Buffer&gt;</code> with the
<code><a href="../../apidocs/io/vertx/rxjava/core/http/HttpClientResponse.html#toObservable--">toObservable</a></code> method:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">request.toObservable().
    subscribe(
        response -&gt; {
          Observable&lt;Buffer&gt; observable = response.toObservable();
          observable.forEach(
              buffer -&gt; {
                // Process buffer
              }
          );
        }
    );</code></pre>
</div>
</div>
<div class="paragraph">
<p>The same flow can be achieved with the <code>flatMap</code> operation:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">request.toObservable().
    flatMap(HttpClientResponse::toObservable).
    forEach(
        buffer -&gt; {
          // Process buffer
        }
    );</code></pre>
</div>
</div>
<div class="paragraph">
<p>We can also unmarshall the <code>Observable&lt;Buffer&gt;</code> into an object using the <code><a href="../../apidocs/io/vertx/rxjava/core/RxHelper.html#unmarshaller-java.lang.Class-">RxHelper.unmarshaller</a></code>
static method. This method creates an <code>Rx.Observable.Operator</code> unmarshalling buffers to an object:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">request.toObservable().
    flatMap(HttpClientResponse::toObservable).
    lift(io.vertx.rxjava.core.RxHelper.unmarshaller(MyPojo.class)).
    forEach(
        pojo -&gt; {
          // Process pojo
        }
    );</code></pre>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_http_server_requests">Http server requests</h3>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rxjava/core/http/HttpServer.html#requestStream--">requestStream</a></code> provides a callback for each incoming
request:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">Observable&lt;HttpServerRequest&gt; requestObservable = server.requestStream().toObservable();
requestObservable.subscribe(request -&gt; {
  // Process request
});</code></pre>
</div>
</div>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/core/http/HttpServerRequest.html">HttpServerRequest</a></code> can then be adapted to an <code>Observable&lt;Buffer&gt;</code>:</p>
</div>
<div class="listingblock">
<div class="content">
<pre>Observable&lt;HttpServerRequest&gt; requestObservable = server.requestStream().toObservable();
requestObservable.subscribe(request -&gt; {
  Observable&lt;Buffer&gt; observable = request.toObservable();
});</pre>
</div>
</div>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rxjava/core/RxHelper.html#unmarshaller-java.lang.Class-">RxHelper.unmarshaller</a></code> can be used to parse and map
a json request to an object:</p>
</div>
<div class="listingblock">
<div class="content">
<pre>Observable&lt;HttpServerRequest&gt; requestObservable = server.requestStream().toObservable();
requestObservable.subscribe(request -&gt; {
  Observable&lt;MyPojo&gt; observable = request.
      toObservable().
      lift(io.vertx.rxjava.core.RxHelper.unmarshaller(MyPojo.class));
});</pre>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_websocket_client">Websocket client</h3>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rxjava/core/http/HttpClient.html#websocketStream-int-java.lang.String-java.lang.String-">websocketStream</a></code> provides a single callback when the websocket
connects, otherwise a failure:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">HttpClient client = vertx.createHttpClient(new HttpClientOptions());
WebSocketStream stream = client.websocketStream(8080, "localhost", "/the_uri");
stream.toObservable().subscribe(
    ws -&gt; {
      // Use the websocket
    },
    error -&gt; {
      // Could not connect
    }
);</code></pre>
</div>
</div>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rxjava/core/http/WebSocket.html">WebSocket</a></code> can then be turned into an <code>Observable&lt;Buffer&gt;</code> easily:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">socketObservable.subscribe(
    socket -&gt; {
      Observable&lt;Buffer&gt; dataObs = socket.toObservable();
      dataObs.subscribe(buffer -&gt; {
        System.out.println("Got message " + buffer.toString("UTF-8"));
      });
    }
);</code></pre>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_websocket_server">Websocket server</h3>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/rxjava/core/http/HttpServer.html#websocketStream--">websocketStream</a></code> provides a callback for each incoming
connection:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">Observable&lt;ServerWebSocket&gt; socketObservable = server.websocketStream().toObservable();
socketObservable.subscribe(
    socket -&gt; System.out.println("Web socket connect"),
    failure -&gt; System.out.println("Should never be called"),
    () -&gt; {
      System.out.println("Subscription ended or server closed");
    }
);</code></pre>
</div>
</div>
<div class="paragraph">
<p>The <code><a href="../../apidocs/io/vertx/core/http/ServerWebSocket.html">ServerWebSocket</a></code> can be turned into an <code>Observable&lt;Buffer&gt;</code> easily:</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="prettyprint highlight"><code class="language-java" data-lang="java">socketObservable.subscribe(
    socket -&gt; {
      Observable&lt;Buffer&gt; dataObs = socket.toObservable();
      dataObs.subscribe(buffer -&gt; {
        System.out.println("Got message " + buffer.toString("UTF-8"));
      });
    }
);</code></pre>
</div>
</div>
</div>
</div>
</div>

        

        
          <div id="footer">
            <div id="footer-text">
              
                Last updated 2016-09-12 08:38:04 CEST
              
              
            </div>
          </div>
        
      </div>
    </div>
  </div>
</div>

<footer>
  <div class="container">
    <div class="row">
      <div class="col-xs-6 col-sm-3 col-md-3 col-lg-2">
        <h2>Vert.x</h2>
        <ul class="list-unstyled">
          <li><a href="http://vertx.io/">Home</a></li>
          <li><a href="http://vertx.io/download/">Download</a></li>
          <li><a href="http://vertx.io/docs/">Documentation</a></li>
          <li><a href="https://github.com/vert-x3/wiki/wiki">Wiki</a></li>
          <li><a href="http://vertx.io/blog/">Blog</a></li>
          <li><a href="http://vertx.io/vertx2/" class="vertx-2-link">Vert.x 2</a></li>
        </ul>
      </div>
      <div class="col-xs-6 col-sm-3 col-md-3 col-lg-2">
        <h2>Community</h2>
        <ul class="list-unstyled">
          <li><a href="http://vertx.io/community/">Help &amp; Contributors</a></li>
          <li><a href="http://vertx.io/materials/">Learning materials</a></li>
          <li><a href="https://groups.google.com/forum/?fromgroups#!forum/vertx">User Group</a></li>
          <li><a href="https://groups.google.com/forum/?fromgroups#!forum/vertx-dev">Developer Group</a></li>
        </ul>
      </div>

      <div class="col-xs-12 col-sm-6 col-lg-offset-2 col-md-6 copyright">
        <p>Vert.x is open source and dual licensed under the <a href="https://www.eclipse.org/org/documents/epl-v10.php">Eclipse Public License 1.0</a> and <a href="https://www.apache.org/licenses/LICENSE-2.0.html">Apache License 2.0</a>.</p>
        <p>This website is licensed under the <a href="http://creativecommons.org/licenses/by-sa/3.0/">CC BY-SA 3.0 License</a>.<br>
        Design by <a href="http://www.michel-kraemer.com">Michel Kr&auml;mer</a>. <a href="http://www.entypo.com">Entypo pictograms</a> by Daniel Bruce.</p>
        <div class="row">
          <div class="col-xs-12 col-lg-offset-1 col-md-5">
            <a href="http://eclipse.org">
            <img class="logo eclipse-logo" src="http://vertx.io/assets/eclipse_logo_grey_small.png" width="204" height="48">
            </a>
          </div>
          <div class="col-xs-12 col-md-offset-2 col-lg-offset-0 col-md-5">
            <a href="http://cloudbees.com">
            <img class="logo cloudbees-logo" src="http://vertx.io/assets/Button-Built-on-CB-1-grey.png" width="180" height="48">
           </a>
          </div>
          <div class="col-xs-12 col-md-offset-2 col-lg-offset-1 col-md-5 jprofiler">
            <a href="http://www.ej-technologies.com/products/jprofiler/overview.html"
            style="text-decoration:none">
            <img class="logo jprofiler-logo" src="http://vertx.io/assets/jprofiler-logo.png" width="48" height="48"><span class="jprofiler-logo">&nbsp; JPROFILER</span>
            </a>
          </div>
        </div>
      </div>
    </div>
  </div>
</footer>

<script src="http://static.jboss.org/theme/js/libs/jquery/jquery-1.9.1.min.js"></script>
<script src="http://vertx.io/javascripts/bootstrap.min.js"></script>
<script src="http://vertx.io/javascripts/highlight.pack.js"></script>
<script>hljs.initHighlightingOnLoad();</script>



<script src="http://vertx.io/javascripts/sidebar.js"></script>


</body>
</html>
